package com.tyc.kafka.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.ThreadPoolExecutor;

/**
 * 类描述
 *
 * @author tyc
 * @version 1.0
 * @date 2022-09-01 11:44:08
 */
@Configuration
public class ThreadPoolConfig {


    /**
     * 配置消费kafka数据的线程池
     */
    @Bean
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor();
        pool.setCorePoolSize(5);
        pool.setMaxPoolSize(20);
        pool.setKeepAliveSeconds(10);
        // 初始化时判断若队列小于0则创建容量为0的SynchronousQueue，避免队列里缓冲数据，这样在系统异常关闭时，就能排除因为阻塞队列丢消息的可能。
        pool.setQueueCapacity(0);
        pool.setThreadNamePrefix("kafka-Consumer-");
        pool.setWaitForTasksToCompleteOnShutdown(true);
        // 由调用它的线程处理任务，多线程处理不过来的时候，能够阻塞在kafka的消费线程上，防止系统崩溃
        pool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        // 默认为false true代表： 线程池关闭的时候等待所有任务都完成后，再继续销毁其他的 Bean
        pool.setWaitForTasksToCompleteOnShutdown(true);
        // 等待时长
        pool.setAwaitTerminationSeconds(30);
        pool.initialize();
        return pool;
    }
}
